#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | cst_ds }}'
name='有发未到问题件'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "运单号\t扫描类型\t扫描时间\t扫描网点\t问题件问题类型\t">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 8 --executor-memory 12g --executor-cores 6  --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=3g -S -e "
    --中心发件信息
  with tmp_centersend as (
    select * from (
      select
      waybill_no,
      max(package_code) as package_code
      from jms_dwd.dwd_tab_barscan_centersend_base_dt
      where dt >= date_add('{{ execution_date |  cst_ds }}',-21)
        and dt <= date_add('{{ execution_date |  cst_ds }}',1)
        and waybill_no not like 'B%'
      group by waybill_no
    ) tmp where package_code is null
  ),

    --中心发件&中心到件，获取运单/站点维度 发件时间、到件时间
  tmp_all as (
  select
     waybill_no
    ,scan_site_code
    ,max(case when type = 1 then scan_time else null end) as send_time
    ,max(case when type = 2 then scan_time else null end) as arrive_time
  from (
    select
         tmp.waybill_no
        ,tmp.scan_site_code
        ,tmp.scan_time
        ,1 as type
    from (
      select
         waybill_no
        ,scan_site_code
        ,scan_time
      from jms_dwd.dwd_tab_barscan_centersend_base_dt
      where dt >= date_add('{{ execution_date |  cst_ds }}',-21)
        and dt <= date_add('{{ execution_date |  cst_ds }}',1)
    ) tmp join tmp_centersend t2 on tmp.waybill_no=t2.waybill_no
  union all
    select
       waybill_no
      ,scan_site_code
      ,scan_time
      ,2 as type
    from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
    where dt >= date_add('{{ execution_date |  cst_ds }}',-21)
      and dt <= date_add('{{ execution_date |  cst_ds }}',1)
  ) tmp2 group by waybill_no,scan_site_code
  ),

    --问题件
    --1、有发未到  2、所有中心未建包T-1
  tmp_difficult as (
  select
     t1.waybill_no
    ,t1.scan_time              --问题件扫描时间
    ,t1.scan_site_code         --问题件扫描网点code
    ,t1.scan_site              --问题件扫描网点名称
    ,t1.problem_type_code      --问题件问题类型
    ,t1.network_type           --扫描网点类型
    ,t1.scan_type                 --操作类型
  from jms_dwd.dwd_tab_barscan_difficult_base_dt t1 --问题件
  left join tmp_centersend t2
         on t1.waybill_no = t2.waybill_no
      where t1.dt = '{{ execution_date |  cst_ds }}'
        and t1.problem_type_code in ('29','29a') --有发未到
        and t1.network_type = 2
        and t2.waybill_no is not null            --所有中心未建包
  )

  select
      t1.waybill_no
     ,t1.scan_type
     ,t1.scan_time
     ,t1.scan_site
     ,problem_type_code
  from tmp_difficult t1
  left join tmp_all t2
         on t1.waybill_no = t2.waybill_no
        and t1.scan_site_code = t2.scan_site_code
      where (unix_timestamp(nvl(t2.send_time,t2.arrive_time)) - unix_timestamp(t1.scan_time))/60/60 < 4
        and (unix_timestamp(nvl(t2.send_time,t2.arrive_time)) - unix_timestamp(t1.scan_time))/60/60 > 0
        and nvl(t2.send_time,t2.arrive_time) is not null
  ;
 " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt